package com.ehotting.mqbd.core.service;

import com.ehotting.eaf.redis.service.RedisService;
import com.ehotting.mqbd.api.MQDBConstants;
import com.ehotting.mqbd.api.bean.ConsumerNodeBean;
import com.ehotting.mqbd.api.enums.ConsumerNodeStateEnum;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Comparator;
import java.util.List;

@Component
public class ConsumerNodeService {

    @Autowired
    private RedisService redisService;
    @Autowired
    private AgentRefConsumerService agentRefConsumerService;

    public ConsumerNodeBean get(String ip){
        List<ConsumerNodeBean> list = listAll();
        return list.stream()
                .filter(t->{return t.getIp().equals(ip);})
                .findFirst()
                .orElse(null);
    }

    public List<ConsumerNodeBean>  listAll(){
        Collection<String> keys = redisService.keys(MQDBConstants.CONSUMER_NODE + "*");
        List<ConsumerNodeBean> list = new ArrayList<ConsumerNodeBean>();
        for(String key : keys){
            ConsumerNodeBean consumerNode = redisService.getCacheObject(key);
            list.add(consumerNode);
        }
        return list;
    }

    public void save(ConsumerNodeBean consumerNode){
        ConsumerNodeBean cn = get(consumerNode.getIp());
        if(cn!=null){
            consumerNode.setCount(cn.getCount());
        }
        redisService.setCacheObject(MQDBConstants.CONSUMER_NODE + consumerNode.getIp(),consumerNode);
    }

    public void delete(String ip){
        agentRefConsumerService.deleteAllAgentByConsumerIp(ip);
        redisService.deleteObject(MQDBConstants.CONSUMER_NODE + ip);
    }

    public void addCount(String ip){
        ConsumerNodeBean consumerNode = get(ip);
        if(consumerNode!=null){
            consumerNode.setCount(consumerNode.getCount()+1);
            redisService.setCacheObject(MQDBConstants.CONSUMER_NODE + consumerNode.getIp(),consumerNode);
        }
    }

    public ConsumerNodeBean getLive(){
        List<ConsumerNodeBean>  list = listAll();
        ConsumerNodeBean consumerNode =  list.stream()
                .filter(t->{return  t.getState() == ConsumerNodeStateEnum.YES;})
                .sorted(Comparator.comparing(ConsumerNodeBean::getCount))
                .findFirst()
                .orElse(null);
        if(consumerNode!=null){
            addCount(consumerNode.getIp());
        }
        return consumerNode;
    }




}
